package net.opentsdb.tsd;

import net.opentsdb.core.TSDB;
import org.jboss.netty.buffer.ChannelBuffer;
import org.jboss.netty.channel.*;
import org.jboss.netty.handler.codec.frame.FrameDecoder;
import org.jboss.netty.handler.codec.frame.LineBasedFrameDecoder;
import org.jboss.netty.handler.codec.http.*;
import org.jboss.netty.handler.codec.string.StringEncoder;
import org.jboss.netty.handler.timeout.IdleStateHandler;
import org.jboss.netty.util.Timer;

import static org.jboss.netty.channel.Channels.pipeline;

/**
 * @author Hujf
 * @title: PipelineFactory
 * @date 2021-09-14 16:20
 * @description: 管道工厂  用于netty handler 管理
 */
public class PipelineFactory implements ChannelPipelineFactory {
    private TSDB tsdb;
    private Timer timer;
    private final ChannelHandler timeoutHandler;
    /**
     * The server side socket timeout.
     **/
    private final int socketTimeout;
    private final ConnectionManager connmgr;
    private final DetectHttpOrRpc HTTP_OR_RPC = new DetectHttpOrRpc();


    // Those are entirely stateless and thus a single instance is needed.
    private static final StringEncoder ENCODER = new StringEncoder();
    private static final WordSplitter DECODER = new WordSplitter();

    /**
     * Stateless handler for RPCs.
     */
    private final RpcHandler rpchandler;

    public PipelineFactory(TSDB tsdb, RpcManager manager, int connections_limit) {
        this.tsdb = tsdb;
        socketTimeout = tsdb.getConfig().getInt("tsd.core.socket.timeout");
        timer = tsdb.getTimer();
        timeoutHandler = new IdleStateHandler(timer, 0, 0, socketTimeout);
        //存放了2个处理器
        rpchandler = new RpcHandler(tsdb, manager);
        connmgr = new ConnectionManager(connections_limit);
    }

    @Override
    public ChannelPipeline getPipeline() throws Exception {
        final ChannelPipeline pipeline = pipeline();

        pipeline.addLast("connmgr", connmgr);
        pipeline.addLast("detect", HTTP_OR_RPC);
        return pipeline;
    }

    /**
     * Dynamically changes the {@link ChannelPipeline} based on the request.
     * If a request uses HTTP, then this changes the pipeline to process HTTP.
     * Otherwise, the pipeline is changed to processes an RPC.
     */
    final class DetectHttpOrRpc extends FrameDecoder {

        @Override
        protected Object decode(final ChannelHandlerContext ctx,
                                final Channel chan,
                                final ChannelBuffer buffer) throws Exception {
            if (buffer.readableBytes() < 1) {  // Yes sometimes we can be called
                return null;                     // with an empty buffer...
            }

            final int firstbyte = buffer.getUnsignedByte(buffer.readerIndex());
            final ChannelPipeline pipeline = ctx.getPipeline();
            // None of the commands in the RPC protocol start with a capital ASCII
            // letter for the time being, and all HTTP commands do (GET, POST, etc.)
            // so use this as a cheap way to differentiate the two.
            if ('A' <= firstbyte && firstbyte <= 'Z') {
                pipeline.addLast("decoder", new HttpRequestDecoder());
                if (tsdb.getConfig().enable_chunked_requests()) {
                    pipeline.addLast("aggregator", new HttpChunkAggregator(
                            tsdb.getConfig().max_chunked_requests()));
                }
                // allow client to encode the payload (ie : with gziped json)
                pipeline.addLast("inflater", new HttpContentDecompressor());
                pipeline.addLast("encoder", new HttpResponseEncoder());
                pipeline.addLast("deflater", new HttpContentCompressor());
            } else {
                pipeline.addLast("framer", new LineBasedFrameDecoder(1024));
                pipeline.addLast("encoder", ENCODER);
                pipeline.addLast("decoder", DECODER);
            }
//
//            if (tsdb.getAuth() != null) {
//                pipeline.addLast("authentication", new AuthenticationChannelHandler(tsdb));
//            }

            pipeline.addLast("timeout", timeoutHandler);
            pipeline.remove(this);
            pipeline.addLast("handler", rpchandler);

            // Forward the buffer to the next handler.
            return buffer.readBytes(buffer.readableBytes());
        }

    }
}
